package net.vinote.sosa.core.rmi;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.math.NumberUtils;

import net.vinote.sosa.rpc.RpcConfig;
import net.vinote.sosa.rpc.RpcResponse;
import net.vinote.sosa.rpc.RpcServiceCallback;
import net.vinote.sosa.rpc.RpcUnit;
import net.vinote.sosa.rpc.Session;
import net.vinote.sosa.rpc.channel.ClientChannelService;
import net.vinote.sosa.rpc.netty.client.RpcNettyClient;
import net.vinote.sosa.rpc.process.DisruptorProcessor;

public class RmiClient {
	private Map<String, ClientChannelService> sessionMap = new ConcurrentHashMap<String, ClientChannelService>();
	private DisruptorProcessor<RpcResponse> processor = new DisruptorProcessor<RpcResponse>();

	public void init() throws Exception {
		RpcConfig config = new RpcConfig();
		config.setRpcCallback(new RpcServiceCallback() {
			@Override
			public <T> void callback(T t) throws Exception {
				@SuppressWarnings("unchecked")
				RpcUnit<RpcResponse> unit = (RpcUnit<RpcResponse>) t;
				if (!unit.getSession().notifySyncMessage(unit.getUnit())) {
					//System.out.println(Thread.currentThread()+"  Ingore Message " + unit.getUnit());
				}
			}
		});
		processor.init(config);
	}

	public Session getSession(String url) {
		if (sessionMap.containsKey(url)) {
			return sessionMap.get(url).getSession();
		}
		synchronized (url) {
			if (sessionMap.containsKey(url)) {
				return sessionMap.get(url).getSession();
			}
			try {
				ClientChannelService client = connectAndAuth(url);
				sessionMap.put(url, client);
				return client.getSession();
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	}

	private ClientChannelService connectAndAuth(String url) throws Exception {
		String[] array = url.split(":");
		ClientChannelService client = new RpcNettyClient();
		//ClientChannelService client = new RpcSocketClient();
		RpcConfig config = new RpcConfig();
		config.setHost(array[0]);
		config.setPort(NumberUtils.toInt(array[1]));
		client.start(config, processor);
		return client;
	}

	public void destory() {
		if (MapUtils.isEmpty(sessionMap)) {
			return;
		}
		for (ClientChannelService unit : sessionMap.values()) {
			unit.shutdown();
		}
		processor.shutdown();
		sessionMap.clear();
	}
}
